package com.k.activemq.mqtt.demo1;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

/**
 * 如果使用 ActiveMQ 需要修改配置支持MQTT
 * D:\apache\apache-activemq\apache-activemq-5.14.5\conf
 * case:pass
 *
 * @author kivil
 */
public class MqttListener1 {
    public static void main(String[] args) {
        try{
            String url = "tcp://127.0.0.1:1883";
            String clientId = "client-subscribe-101";
            String username = "admin";
            String password="admin";
            MqttClient client = new MqttClient(url,clientId,new MemoryPersistence());
            client.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable throwable) {
                    System.out.println(clientId+"与服务器断开连接");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println(String.format("接收消息主题 : %s",topic));
                    System.out.println(String.format("接收消息Qos : %d",message.getQos()));
                    System.out.println(String.format("接收消息内容 : %s",new String(message.getPayload())));
                    System.out.println(String.format("接收消息retained : %b",message.isRetained()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
                }
            });
            MqttConnectOptions options = new MqttConnectOptions();
            //是否清空session，设置false表示服务器会保留客户端的连接记录（订阅主题，qos）,客户端重连之后能获取到服务器在客户端断开连接期间推送的消息
            //设置为true表示每次连接服务器都是以新的身份
            options.setCleanSession(true);
            //设置连接用户名
            options.setUserName(username);
            //设置连接密码
            options.setPassword(password.toCharArray());
            //设置超时时间，单位为秒
            options.setConnectionTimeout(100);
            //设置心跳时间 单位为秒，表示服务器每隔 1.5*20秒的时间向客户端发送心跳判断客户端是否在线
            options.setKeepAliveInterval(20);
            //设置遗嘱消息的话题，若客户端和服务器之间的连接意外断开，服务器将发布客户端的遗嘱信息
            options.setWill("willTopic",(clientId + "与服务器断开连接").getBytes(),0,false);
            client.connect(options);

            // subscribe
            //订阅主题
            //消息等级，和主题数组一一对应，服务端将按照指定等级给订阅了主题的客户端推送消息
            int[] qos = {0};
            String[] topics = {"mqtt-topic-101/a"};
            subscribe(client,topics,qos);
        }catch (Exception e){
            e.printStackTrace();
        }
    }
    public static void subscribe(MqttClient client,String[] topics,int[] qos) throws MqttException {
        client.subscribe(topics,qos);
    }

}
